package org.hgf.testrabbitmq.producer.service.impl;

import cn.hutool.core.thread.ThreadUtil;
import org.hgf.testrabbitmq.consumer.config.ConsumerRabbitConfig;
import org.hgf.testrabbitmq.consumer.service.IRabbitMqConsumerService;
import org.hgf.testrabbitmq.domain.MqConfig;
import org.hgf.testrabbitmq.producer.service.IRabbitMqProducerService;
import org.hgf.testrabbitmq.utils.enums.ExchangeTypeEnum;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;

/**
 * rabbitmq服务具体实现（生产者）
 *
 * @author 何国发
 * @version 1.0.0
 * @date 2022/9/7 15:04
 */
@Service
public class RabbitMqProducerServiceImpl implements IRabbitMqProducerService {

    /** 线程池 */
    private static final ExecutorService EXECUTOR = ThreadUtil.newExecutor();

    /**
     * rabbit管理辅助操作类
     */
    @Resource
    private RabbitAdmin rabbitAdmin;

    /**
     * rabbitmq服务（消费者）
     */
    @Resource
    private IRabbitMqConsumerService consumerService;

    /**
     * 如果不存在原有队列就创建一个新队列
     *
     * @param mqConfig 消息队列相关配置
     * @author 何国发
     * @date 2022/9/7 15:02
     */
    @Override
    public void createQueueIfNotExist(MqConfig mqConfig) {
        // 参数校验
        this.checkParam(mqConfig);

        // 根据队列名称，查看队列是否存在
        Properties queueProperties = rabbitAdmin.getQueueProperties(mqConfig.getQueueName());

        // 不存在就创建一个新的队列
        if (queueProperties == null)
        {
            // 将创建的队列与死信交换机绑定
            Map<String, Object> paramMap = new HashMap<>(15);
            // 绑定死信交换机
            paramMap.put("x-dead-letter-exchange", ConsumerRabbitConfig.DEAD_EXCHANGE_NAME);

            // 创建队列
            Queue queue = new Queue(mqConfig.getQueueName(), true, false, false, paramMap);

            // 根据传入的类型构建交换机(默认构建扇出交换机)
            Exchange exchange = this.buildExchange(mqConfig.getExchangeTypeEnum(), mqConfig.getExchangeName());

            // 进行具体构建
            rabbitAdmin.declareQueue(queue);
            rabbitAdmin.declareExchange(exchange);
            rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(mqConfig.getRouteKey()).and(paramMap));

            // 动态增加一个监听器
            consumerService.addNewListener(mqConfig.getQueueName());

            // 新开一个线程新增消息监听器（线程异步会导致监听队列还没创建就发送消息了）
//            EXECUTOR.execute(() -> consumerService.addNewListener(mqConfig.getQueueName()));

        }
    }

    /**
     * 校验参数
     *
     * @param mqConfig  消息队列相关配置
     * @author 何国发
     * @date 2022/9/7 15:41
     */
    private void checkParam(MqConfig mqConfig) {
        if (mqConfig.getQueueName() == null || "".equals(mqConfig.getQueueName()))
        {
            throw new RuntimeException("创建失败，队列名称为空！");
        }
        if (mqConfig.getExchangeName() == null || "".equals(mqConfig.getExchangeName()))
        {
            throw new RuntimeException("创建失败，队列名称为空！");
        }
    }

    /**
     * 根据传入的类型构建交换机(默认构建扇出交换机)
     *
     * @param exchangeTypeEnum  交换机类型枚举
     * @param exchangeName      交换机名称
     * @return org.springframework.amqp.core.Exchange
     * @author 何国发
     * @date 2022/9/7 15:35
     */
    private Exchange buildExchange(ExchangeTypeEnum exchangeTypeEnum, String exchangeName)
    {
        Exchange exchange;
        switch (exchangeTypeEnum)
        {
            case DIRECT:
                exchange = new DirectExchange(exchangeName);
                break;
            case TOPIC:
                exchange = new TopicExchange(exchangeName);
                break;
            case HEADERS:
                exchange = new HeadersExchange(exchangeName);
                break;
            default:
                exchange = new FanoutExchange(exchangeName);
        }

        return exchange;
    }
}
